package com.rxjava;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;

//Catch类似于java 中的try/catch，拦截onError的调用，让Observable不会因为错误的产生而终止。
public class CatchDemo {

	private Observable<String> createObserver(Boolean createExcetion) {
        return Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) {
                for (int i = 1; i <= 6; i++) {
                    if (i < 3) {
                    	emitter.onNext("onNext:" + i);
                    } else if (createExcetion) {
                    	emitter.onError(new Exception("Exception"));
                    } else {
                    	emitter.onError(new Throwable("Throw error"));
                    }
                }
            }
        });
    }

	public void onErrorReturn() {
		//当发生错误的时候，让Observable发射一个预先定义好的数据并正常地终止
		//onErrorReturn方法 返回一个镜像原有Observable行为的新Observable
		//会忽略前者的onError调用，不会将错误传递给观察者，而是发射一个特殊的项并调用观察者的onCompleted方法。
        createObserver(true).onErrorReturn(throwable -> "onErrorReturn：" + throwable.getMessage())
        .subscribe(new Observer<String>() {
            @Override
            public void onComplete() {
                System.out.println("onErrorReturn-onCompleted\n");
            }

            @Override
            public void onError(Throwable e) {
            	System.out.println("onErrorReturn-onError:" + e.getMessage());
            }

            @Override
            public void onNext(String s) {
            	System.out.println("onErrorReturn-onNext:" + s);
            }

			@Override
			public void onSubscribe(Disposable disposable) {

			}
        });
    }

	public void onErrorResumeNext() {
		//OnErrorResume
		//当发生错误的时候，由另外一个Observable来代替当前的Observable并继续发射数据
		//onErrorResumeNext方法返回一个镜像原有Observable行为的新Observable
        createObserver(false).onErrorResumeNext(Observable.just("7", "8", "9"))
        .subscribe(new Observer<String>() {
            @Override
            public void onComplete() {
            	System.out.println("onErrorResume-onCompleted\n");
            }

            @Override
            public void onError(Throwable e) {
            	System.out.println("onErrorResume-onError:" + e.getMessage());
            }

            @Override
            public void onNext(String s) {
            	System.out.println("onErrorResume-onNext:" + s);
            }

			@Override
			public void onSubscribe(Disposable disposable) {

			}
        });
    }

	public void onExceptionResumeTrue() {
		//OnExceptionResumeNext
		//类似于OnErrorResume,不同之处在于其会对onError抛出的数据类型做判断，
		//如果是Exception，也会使用另外一个Observable代替原Observable继续发射数据，否则会将错误分发给Subscriber。
		createObserver(false).onExceptionResumeNext(Observable.just("7", "8", "9"))
		.subscribe(new Observer<String>() {
            @Override
            public void onComplete() {
            	System.out.println("onException-true-onCompleted\n");
            }

            @Override
            public void onError(Throwable e) {
            	System.out.println("onException-true-onError:" + e.getMessage());
            }

            @Override
            public void onNext(String s) {
            	System.out.println("onException-true-onNext:" + s);
            }

			@Override
			public void onSubscribe(Disposable disposable) {

			}
        });
    }

	public static void main(String[] args) {
		CatchDemo demo = new CatchDemo();
		//demo.onErrorReturn();
		//demo.onErrorResumeNext();
		demo.onExceptionResumeTrue();
	}

}
